package com.atguigu.edu.realtime.utils;

import com.atguigu.edu.realtime.common.GmallConfig;
import org.apache.commons.lang3.StringUtils;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.sink.DorisSink;
import org.apache.doris.flink.sink.writer.SimpleStringSerializer;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;


import java.util.List;
import java.util.Properties;

public class DorisUtil {

    public static DorisSink<String> getDorisSink(String table) {
        Properties props = new Properties();
        props.setProperty("format", "json");
        props.setProperty("read_json_by_line", "true"); // 每行一条 json 数据

        return DorisSink.<String>builder()
                .setDorisReadOptions(DorisReadOptions.builder().build())
                .setDorisOptions(DorisOptions.builder() // 设置 doris 的连接参数
                        .setFenodes(GmallConfig.DORIS_FE)
                        .setTableIdentifier(GmallConfig.DORIS_DB + "." + table)
                        .setUsername("root")
                        .setPassword("aaaaaa")
                        .build())
                .setDorisExecutionOptions(DorisExecutionOptions.builder() // 执行参数
                        //.setLabelPrefix("doris-label")  // stream-load 导入的时候的 label 前缀
                        //.disable2PC() // 开启两阶段提交后,labelPrefix 需要全局唯一,为了测试方便禁用两阶段提交
                        .setDeletable(false)
                        .setBufferCount(3) // 批次条数: 默认 3
                        .setBufferSize(8 * 1024) // 批次大小: 默认 1M
                        .setCheckInterval(3000) // 批次输出间隔   三个对批次的限制是或的关系
                        .setMaxRetries(3)
                        .setStreamLoadProp(props) // 设置 stream load 的数据格式 默认是 csv,需要改成 json
                        .build())
                .setSerializer(new SimpleStringSerializer())
                .build();
    }



        private static final String JDBC_DRIVER = "com.mysql.cj.jdbc.Driver";
        private static final String DB_URL_PATTERN = "jdbc:mysql://hadoop102:9030/edu?rewriteBatchedStatements=true";
        private static final String HOST = "hadoop102"; // Leader Node host
        private static final int PORT = 9030;   // query_port of Leader Node
        private static final String DB = "edu";
        private static  String TBL;//表名
        private static final String USER = "root";
        private static final String PASSWD = "aaaaaa";
        private static String SQL;

        private static final int INSERT_BATCH_SIZE = 100;


        //向doris插入数据
    public static void insertDoris(Connection conn , String Tbl , List values) {
        // 注意末尾不要加 分号 ";"
        TBL = Tbl;
        if (StringUtils.isEmpty(SQL)){
            String sql = "insert into " + TBL +" values(";//?, ?)";
            for (int i = 0; i < values.size(); i++) {
                if (i!=values.size()-1){
                    sql = sql + "?,";
                }else {
                    sql = sql + "?";
                }
            }
            sql =sql + ")";
            SQL=sql;

        }

        PreparedStatement stmt = null;
        try {
            stmt = conn.prepareStatement(SQL);

            int i=1;
            for (Object value : values) {
                stmt.setObject(i,value);
                i++;
            }
            stmt.addBatch();
            int[] res = stmt.executeBatch();

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (stmt != null) {
                    stmt.close();
                }
            } catch (SQLException se2) {
                se2.printStackTrace();
            }

        }
    }
    //关闭jdbc连接
    public static void closeJdbcConnection (Connection conn){
        try {
            if (conn != null) conn.close();
        } catch (SQLException se) {
            se.printStackTrace();
        }
    }
        //获取jdbc连接
        public static Connection getJdbcConnection () {

            Connection conn = null;
            String dbUrl = String.format(DB_URL_PATTERN, HOST, PORT, DB);
            try {
                Class.forName(JDBC_DRIVER);
                conn = DriverManager.getConnection(dbUrl, USER, PASSWD);

            } catch (Exception e) {
                e.printStackTrace();
            }
            return conn;

        }



}

